package com.chb.flink.window

import com.chb.flink.source.{MyCustomerSource}
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time

object TestReduceFunctionByWindow {

    //每隔5秒统计每个基站的日志数量
    def main(args: Array[String]): Unit = {
        val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
        import org.apache.flink.streaming.api.scala._

        val data = streamEnv.addSource(new MyCustomerSource)
        // 10:05~10:10  10:10~10:15   10.08


        //开窗
        data.map(log => (log.sid, 1)).keyBy(_._1)
            .timeWindow(Time.seconds(5))
            .reduce((t1, t2) => (t1._1, t1._2 + t2._2))
            .print()

        streamEnv.execute()
    }
}
